package com.itheima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.itheima.behavior.config.KafkaConfig;
import com.itheima.behavior.service.BehaviorTrackService;
import com.itheima.common.redis.CacheService;
import com.itheima.model.behavior.dtos.BehaviorDto;
import com.itheima.model.behavior.pojos.ReadingStatistics;
import com.itheima.model.behavior.pojos.UserBehavior;
import com.itheima.model.behavior.vos.BrowsingHistoryVo;
import com.itheima.model.common.dtos.PageResponseResult;
import com.itheima.model.common.dtos.ResponseResult;
import com.itheima.utils.thread.MpThreadLocalUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 行为追踪服务实现类（使用 MongoTemplate）
 */
@Service
@Slf4j
public class BehaviorTrackServiceImpl implements BehaviorTrackService {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private CacheService cacheService;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static final String READING_SESSION_KEY = "reading_session:%s:%s"; // userId:bookId
    private static final String DAILY_STATS_KEY = "user_stats:%s:%s"; // userId:date

    /**
     * 单条行为上报
     */
    @Override
    public ResponseResult trackBehavior(BehaviorDto dto) {
        try {
            Integer userId = MpThreadLocalUtil.getUser().getId();

            // 构建行为记录
            UserBehavior behavior = UserBehavior.builder()
                    .userId(userId)
                    .behaviorType(dto.getBehaviorType())
                    .targetType(dto.getTargetType())
                    .targetId(dto.getTargetId())
                    .duration(dto.getDuration() != null ? dto.getDuration() : 0)
                    .behaviorData(dto.getExtraData())
                    .createTime(new Date())
                    .build();

            // 保存到 MongoDB
            mongoTemplate.save(behavior);

            // 发送事件到 Kafka（用于流式计算排行榜）
            sendBehaviorEventToKafka(behavior);

            // 实时更新 Redis 缓存（用于实时统计）
            updateRealtimeCache(userId, dto);

            log.info("行为上报成功: userId={}, type={}, target={}:{}",
                    userId, dto.getBehaviorType(), dto.getTargetType(), dto.getTargetId());

            return ResponseResult.okResult("行为上报成功");

        } catch (Exception e) {
            log.error("行为上报失败", e);
            return ResponseResult.errorResult(500, "行为上报失败");
        }
    }

    /**
     * 批量行为上报
     */
    @Override
    public ResponseResult trackBehaviorBatch(List<BehaviorDto> dtoList) {
        try {
            Integer userId = MpThreadLocalUtil.getUser().getId();

            // 构建行为记录列表
            List<UserBehavior> behaviors = dtoList.stream()
                    .map(dto -> UserBehavior.builder()
                            .userId(userId)
                            .behaviorType(dto.getBehaviorType())
                            .targetType(dto.getTargetType())
                            .targetId(dto.getTargetId())
                            .duration(dto.getDuration() != null ? dto.getDuration() : 0)
                            .behaviorData(dto.getExtraData())
                            .createTime(new Date())
                            .build())
                    .collect(Collectors.toList());

            // 批量保存到 MongoDB
            mongoTemplate.insertAll(behaviors);

            // 批量发送事件到 Kafka
            behaviors.forEach(this::sendBehaviorEventToKafka);

            // 更新缓存
            dtoList.forEach(dto -> updateRealtimeCache(userId, dto));

            log.info("批量行为上报成功: userId={}, count={}", userId, dtoList.size());

            return ResponseResult.okResult("批量行为上报成功");

        } catch (Exception e) {
            log.error("批量行为上报失败", e);
            return ResponseResult.errorResult(500, "批量行为上报失败");
        }
    }

    /**
     * 开始阅读
     */
    @Override
    public ResponseResult startReading(Integer bookId) {
        try {
            Integer userId = MpThreadLocalUtil.getUser().getId();
            String sessionKey = String.format(READING_SESSION_KEY, userId, bookId);

            // 记录开始时间到 Redis
            Map<String, Object> session = new HashMap<>();
            session.put("startTime", System.currentTimeMillis());
            session.put("bookId", bookId);
            session.put("userId", userId);

            cacheService.setEx(sessionKey, JSON.toJSONString(session), 7200, TimeUnit.SECONDS); // 2小时过期

            // 记录行为
            trackBehavior(BehaviorDto.builder()
                    .behaviorType("reading_start")
                    .targetType("book")
                    .targetId(bookId.toString())
                    .build());

            log.info("开始阅读: userId={}, bookId={}", userId, bookId);

            return ResponseResult.okResult("开始阅读记录成功");

        } catch (Exception e) {
            log.error("开始阅读记录失败", e);
            return ResponseResult.errorResult(500, "开始阅读记录失败");
        }
    }

    /**
     * 结束阅读
     */
    @Override
    public ResponseResult endReading(Integer bookId, Integer duration) {
        try {
            Integer userId = MpThreadLocalUtil.getUser().getId();
            String sessionKey = String.format(READING_SESSION_KEY, userId, bookId);

            // 从 Redis 获取开始时间
            String sessionJson = cacheService.get(sessionKey);
            if (sessionJson != null) {
                // 计算实际阅读时长
                Map session = JSON.parseObject(sessionJson, Map.class);
                long startTime = ((Number) session.get("startTime")).longValue();
                long actualDuration = (System.currentTimeMillis() - startTime) / 1000;

                // 使用前端传的时长和实际时长的较小值（防止作弊）
                duration = (int) Math.min(duration, actualDuration);
            }

            // 记录行为
            trackBehavior(BehaviorDto.builder()
                    .behaviorType("reading_end")
                    .targetType("book")
                    .targetId(bookId.toString())
                    .duration(duration)
                    .build());

            // 删除会话
            cacheService.delete(sessionKey);

            // 更新当日阅读统计
            updateDailyStatistics(userId, duration);

            log.info("结束阅读: userId={}, bookId={}, duration={}s", userId, bookId, duration);

            Map<String, Object> result = new HashMap<>();
            result.put("duration", duration);
            result.put("minutes", duration / 60);

            return ResponseResult.okResult(result);

        } catch (Exception e) {
            log.error("结束阅读记录失败", e);
            return ResponseResult.errorResult(500, "结束阅读记录失败");
        }
    }

    /**
     * 获取用户行为统计
     */
    @Override
    public ResponseResult getUserBehaviorStats(Integer userId) {
        try {
            // 获取今日统计
            LocalDate today = LocalDate.now();
            Query todayQuery = new Query(Criteria.where("user_id").is(userId).and("stat_date").is(today));
            ReadingStatistics todayStats = mongoTemplate.findOne(todayQuery, ReadingStatistics.class);
            
            if (todayStats == null) {
                todayStats = ReadingStatistics.builder()
                        .userId(userId)
                        .statDate(today)
                        .readingTime(0)
                        .booksRead(0)
                        .notesCount(0)
                        .reviewsCount(0)
                        .pageViews(0)
                        .build();
            }

            // 获取本周统计
            LocalDate weekStart = today.minusDays(6);
            Query weekQuery = new Query(Criteria.where("user_id").is(userId)
                    .and("stat_date").gte(weekStart).lte(today));
            List<ReadingStatistics> weekStats = mongoTemplate.find(weekQuery, ReadingStatistics.class);

            int weekReadingTime = weekStats.stream()
                    .mapToInt(ReadingStatistics::getReadingTime)
                    .sum();

            int weekBooksRead = weekStats.stream()
                    .mapToInt(ReadingStatistics::getBooksRead)
                    .sum();

            // 统计行为总数
            Query countQuery = new Query(Criteria.where("user_id").is(userId));
            long totalBehaviors = mongoTemplate.count(countQuery, UserBehavior.class);

            Map<String, Object> stats = new HashMap<>();
            stats.put("today", todayStats);
            stats.put("weekReadingTime", weekReadingTime);
            stats.put("weekBooksRead", weekBooksRead);
            stats.put("weekStats", weekStats);
            stats.put("totalBehaviors", totalBehaviors);

            return ResponseResult.okResult(stats);

        } catch (Exception e) {
            log.error("获取用户行为统计失败", e);
            return ResponseResult.errorResult(500, "获取用户行为统计失败");
        }
    }

    /**
     * 发送行为事件到 Kafka
     */
    private void sendBehaviorEventToKafka(UserBehavior behavior) {
        try {
            // 构建事件数据
            Map<String, Object> eventData = new HashMap<>();
            eventData.put("userId", behavior.getUserId());
            eventData.put("behaviorType", behavior.getBehaviorType());
            eventData.put("targetType", behavior.getTargetType());
            eventData.put("targetId", behavior.getTargetId());
            eventData.put("duration", behavior.getDuration());
            eventData.put("createTime", behavior.getCreateTime().getTime());
            
            if (behavior.getBehaviorData() != null) {
                eventData.put("behaviorData", behavior.getBehaviorData());
            }

            String eventJson = JSON.toJSONString(eventData);
            String key = behavior.getTargetType() + ":" + behavior.getTargetId();

            // 发送到 Kafka
            kafkaTemplate.send(KafkaConfig.Topics.USER_BEHAVIOR, key, eventJson)
                    .addCallback(
                            success -> log.debug("发送事件到Kafka成功: topic={}, key={}", 
                                    KafkaConfig.Topics.USER_BEHAVIOR, key),
                            failure -> log.error("发送事件到Kafka失败: topic={}, key={}", 
                                    KafkaConfig.Topics.USER_BEHAVIOR, key, failure)
                    );

        } catch (Exception e) {
            log.error("发送行为事件到Kafka失败", e);
            // 不抛出异常，避免影响主流程
        }
    }

    /**
     * 更新实时缓存
     */
    private void updateRealtimeCache(Integer userId, BehaviorDto dto) {
        try {
            String today = LocalDate.now().toString();
            String cacheKey = String.format(DAILY_STATS_KEY, userId, today);

            // 更新今日统计
            if ("page_view".equals(dto.getBehaviorType())) {
                cacheService.hIncrBy(cacheKey, "page_views", 1);
            }

            if ("reading_end".equals(dto.getBehaviorType()) && dto.getDuration() != null) {
                cacheService.hIncrBy(cacheKey, "reading_time", dto.getDuration() / 60);
            }

            if ("note_create".equals(dto.getBehaviorType())) {
                cacheService.hIncrBy(cacheKey, "notes_count", 1);
            }

            if ("review_create".equals(dto.getBehaviorType())) {
                cacheService.hIncrBy(cacheKey, "reviews_count", 1);
            }

            // 设置过期时间为3天
            cacheService.expire(cacheKey, 259200, TimeUnit.SECONDS);

        } catch (Exception e) {
            log.error("更新实时缓存失败", e);
        }
    }

    /**
     * 更新当日阅读统计（存储到 MongoDB）
     * 使用 upsert 操作，如果不存在则插入，存在则更新
     */
    private void updateDailyStatistics(Integer userId, Integer duration) {
        try {
            LocalDate today = LocalDate.now();

            // 构建查询条件
            Query query = new Query(Criteria.where("user_id").is(userId).and("stat_date").is(today));

            // 构建更新操作
            Update update = new Update()
                    .inc("reading_time", duration / 60)  // 增加阅读时长（分钟）
                    .inc("books_read", 1)                 // 增加阅读图书数
                    .setOnInsert("user_id", userId)       // 插入时设置
                    .setOnInsert("stat_date", today)      // 插入时设置
                    .setOnInsert("notes_count", 0)        // 插入时设置
                    .setOnInsert("reviews_count", 0)      // 插入时设置
                    .setOnInsert("page_views", 0)         // 插入时设置
                    .setOnInsert("create_time", new Date()); // 插入时设置

            // 执行 upsert 操作
            mongoTemplate.upsert(query, update, ReadingStatistics.class);

            log.info("更新当日统计成功: userId={}, date={}, duration={}min",
                    userId, today, duration / 60);

        } catch (Exception e) {
            log.error("更新当日统计失败", e);
        }
    }

    /**
     * 获取用户浏览记录
     */
    @Override
    public PageResponseResult getBrowsingHistory(Integer userId, String timeFilter, Integer page, Integer size) {
        try {
            // 构建查询条件
            Query query = new Query();
            query.addCriteria(Criteria.where("userId").is(userId));
            
            // 浏览类型的行为(只包含内容浏览,不包含页面浏览)
            List<String> viewBehaviorTypes = Arrays.asList("book_view", "note_view", "review_view");
            query.addCriteria(Criteria.where("behaviorType").in(viewBehaviorTypes));
            
            // 只查询内容类型,不包含页面
            List<String> contentTypes = Arrays.asList("book", "note", "review");
            query.addCriteria(Criteria.where("targetType").in(contentTypes));

            // 时间筛选
            Date startTime = getStartTimeByFilter(timeFilter);
            if (startTime != null) {
                query.addCriteria(Criteria.where("createTime").gte(startTime));
            }

            // 排序和分页
            query.with(Sort.by(Sort.Direction.DESC, "createTime"));
            long total = mongoTemplate.count(query, UserBehavior.class);

            query.skip((long) (page - 1) * size);
            query.limit(size);

            List<UserBehavior> behaviors = mongoTemplate.find(query, UserBehavior.class);

            // 转换为 VO
            List<BrowsingHistoryVo> historyList = behaviors.stream()
                    .map(this::convertToBrowsingHistoryVo)
                    .collect(Collectors.toList());

            PageResponseResult result = new PageResponseResult(page, size, (int) total);
            result.setData(historyList);

            return result;

        } catch (Exception e) {
            log.error("获取浏览记录失败", e);
            return new PageResponseResult(page, size, 0);
        }
    }

    /**
     * 根据时间筛选条件获取开始时间
     */
    private Date getStartTimeByFilter(String timeFilter) {
        if (timeFilter == null || "all".equals(timeFilter)) {
            return null;
        }

        LocalDateTime startDateTime = null;
        LocalDate today = LocalDate.now();

        switch (timeFilter) {
            case "today":
                startDateTime = today.atStartOfDay();
                break;
            case "week":
                startDateTime = today.minusDays(today.getDayOfWeek().getValue() - 1).atStartOfDay();
                break;
            case "month":
                startDateTime = today.withDayOfMonth(1).atStartOfDay();
                break;
        }

        return startDateTime != null
                ? Date.from(startDateTime.atZone(ZoneId.systemDefault()).toInstant())
                : null;
    }

    /**
     * 转换为浏览记录VO
     */
    private BrowsingHistoryVo convertToBrowsingHistoryVo(UserBehavior behavior) {
        BrowsingHistoryVo vo = new BrowsingHistoryVo();
        vo.setId(behavior.getId());
        vo.setContentType(behavior.getTargetType());
        vo.setContentId(behavior.getTargetId());
        vo.setDuration(behavior.getDuration());
        vo.setViewTime(behavior.getCreateTime());

        // 从 behaviorData 中提取详细信息
        if (behavior.getBehaviorData() != null) {
            Map<String, Object> behaviorData = behavior.getBehaviorData();
            
            // 获取标题
            if (behaviorData.containsKey("bookTitle")) {
                vo.setTitle((String) behaviorData.get("bookTitle"));
            } else if (behaviorData.containsKey("title")) {
                vo.setTitle((String) behaviorData.get("title"));
            } else {
                vo.setTitle("未知内容");
            }

            // 获取作者
            if (behaviorData.containsKey("author")) {
                vo.setAuthor((String) behaviorData.get("author"));
            }

            // 获取封面
            if (behaviorData.containsKey("cover")) {
                vo.setCover((String) behaviorData.get("cover"));
            }
        } else {
            vo.setTitle("未知内容");
        }

        return vo;
    }

    /**
     * 删除单条浏览记录
     */
    @Override
    public ResponseResult deleteBrowsingRecord(Integer userId, String recordId) {
        try {
            Query query = new Query();
            query.addCriteria(Criteria.where("_id").is(recordId));
            query.addCriteria(Criteria.where("userId").is(userId));

            mongoTemplate.remove(query, UserBehavior.class);

            return ResponseResult.okResult("删除成功");

        } catch (Exception e) {
            log.error("删除浏览记录失败", e);
            return ResponseResult.errorResult(500, "删除失败");
        }
    }

    /**
     * 清空所有浏览记录
     */
    @Override
    public ResponseResult clearAllBrowsingHistory(Integer userId) {
        try {
            Query query = new Query();
            query.addCriteria(Criteria.where("userId").is(userId));
            
            // 浏览类型的行为(只包含内容浏览,不包含页面浏览)
            List<String> viewBehaviorTypes = Arrays.asList("book_view", "note_view", "review_view");
            query.addCriteria(Criteria.where("behaviorType").in(viewBehaviorTypes));
            
            // 只删除内容类型,不包含页面
            List<String> contentTypes = Arrays.asList("book", "note", "review");
            query.addCriteria(Criteria.where("targetType").in(contentTypes));

            mongoTemplate.remove(query, UserBehavior.class);

            return ResponseResult.okResult("清空成功");

        } catch (Exception e) {
            log.error("清空浏览记录失败", e);
            return ResponseResult.errorResult(500, "清空失败");
        }
    }
    
}

